package com.viknix.threadpool.manager.client.core;

import com.alibaba.fastjson.JSON;
import com.viknix.threadpool.manager.client.annotations.ThreadSafe;
import com.viknix.threadpool.manager.client.consistency.CacheData;
import com.viknix.threadpool.manager.client.constant.Constants;
import com.viknix.threadpool.manager.client.constant.PropertyKeyConst;
import com.viknix.threadpool.manager.client.exception.TpmProcessException;
import com.viknix.threadpool.manager.client.filter.ConfigFilterChainManager;
import com.viknix.threadpool.manager.client.http.HttpAgent;
import com.viknix.threadpool.manager.client.listener.Listener;
import com.viknix.threadpool.manager.client.util.ParamUtil;
import com.viknix.threadpool.manager.common.model.PoolParameterInfo;
import com.viknix.threadpool.manager.common.util.ContentUtil;
import com.viknix.threadpool.manager.common.util.GroupKeyUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import com.viknix.threadpool.manager.client.http.HttpSimpleClient.HttpResult;
import org.apache.commons.lang3.math.NumberUtils;

import java.io.IOException;
import java.net.HttpURLConnection;
import java.net.URLDecoder;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static com.viknix.threadpool.manager.common.constant.Constants.*;

/**
 * @Author: Dongqi
 * @Date: 2021/10/25 09:44
 * @Version 1.0
 * @Description: 该对象主要任务是进行长轮询任务，监听服务端配置更新
 */
@Slf4j
public class ClientWorker {

    /**
     * 只有一个核心线程，用于定期检查线程池配置
     */
    private final ScheduledExecutorService executor;

    /**
     * 核心线程数量等于CPU核心数，用于执行长轮询任务
     */
    private final ScheduledExecutorService executorService;

    /**
     * groupKey -> cacheData
     */
    private Map<String, CacheData> cacheMap = new ConcurrentHashMap(16);

    private final HttpAgent agent;
    private final ConfigFilterChainManager configFilterChainManager;
    private boolean isHealthServer = true;
    /**
     * 超时时间
     */
    private long timeout;
    /**
     * 当前正在执行长轮询的线程数量
     */
    private double currentLongingTaskCount = 0;
    /**
     * 任务延时时间，默认2000
     */
    private int taskPenaltyTime;
    /**
     * 是否允许通远程同步配置
     */
    private boolean enableRemoteSyncConfig = false;

    public boolean isHealthServer() {
        return isHealthServer;
    }

    private void setHealthServer(boolean isHealthServer) {
        this.isHealthServer = isHealthServer;
    }

    /**
     * 注册 Listener
     *
     * @param poolId    线程池id
     * @param item      项目
     * @param listeners 监听器
     * @throws TpmProcessException
     */
    public void addTenantListeners(String poolId, String item, List<? extends Listener> listeners) throws TpmProcessException {
        item = null2defaultItem(item);
        String namespace = agent.getNamespace();
        CacheData cache = addCacheDataIfAbsent(poolId, item, namespace);
        for (Listener listener : listeners) {
            cache.addListener(listener);
        }
    }

    /**
     * 注册 Listener 同时设置 content
     *
     * @param poolId    线程池id
     * @param item      项目
     * @param content   配置信息
     * @param listeners 监听器
     * @throws TpmProcessException
     */
    public void addTenantListenersWithContent(String poolId, String item, String content, List<? extends Listener> listeners) throws TpmProcessException {
        item = null2defaultItem(item);
        String namespace = agent.getNamespace();
        CacheData cache = addCacheDataIfAbsent(poolId, item, namespace);
        cache.setContent(content);
        for (Listener listener : listeners) {
            cache.addListener(listener);
        }
    }

    /**
     * 从 cache 中移除 Listener
     *
     * @param poolId   线程池id
     * @param item     项目
     * @param listener 监听器
     */
    public void removeTenantListener(String poolId, String item, Listener listener) {
        item = null2defaultItem(item);
        String namespace = agent.getNamespace();
        CacheData cache = getCache(poolId, item, namespace);
        if (null != cache) {
            cache.removeListener(listener);
            if (cache.getListeners().isEmpty()) {
                removeCache(poolId, item, namespace);
            }
        }
    }

    /**
     * 从 cacheMap 中移除 groupKey 对应的 CacheData
     *
     * @param poolId    线程池id
     * @param item      项目
     * @param namespace 命名空间
     */
    @ThreadSafe
    void removeCache(String poolId, String item, String namespace) {
        String groupKey = GroupKeyUtil.getKey(poolId, item, namespace);
        synchronized (cacheMap) {
            Map<String, CacheData> copy = new HashMap<String, CacheData>(cacheMap);
            copy.remove(groupKey);
            cacheMap = copy;
        }
        log.info("[{}] [unsubscribe] {}", agent.getName(), groupKey);
    }

    /**
     * 向 cacheMap 中添加 CacheData
     *
     * @param poolId    线程池id
     * @param item      项目
     * @param namespace 命名空间
     * @return CacheData
     * @throws TpmProcessException
     */
    @ThreadSafe
    public CacheData addCacheDataIfAbsent(String poolId, String item, String namespace) throws TpmProcessException {
        // synchronized + 双重判断
        CacheData cache = getCache(poolId, item, namespace);
        if (null != cache) {
            return cache;
        }
        String key = GroupKeyUtil.getKey(poolId, item, namespace);
        synchronized (cacheMap) {
            CacheData cacheFromMap = getCache(poolId, item, namespace);
            // multiple listeners on the same poolId+item and race condition,so
            // double check again
            // other listener thread beat me to set to cacheMap
            if (null != cacheFromMap) {
                cache = cacheFromMap;
                // 重置以便服务器不会挂起此检查
                cache.setInitializing(true);
            } else {
                cache = new CacheData(configFilterChainManager, agent.getName(), poolId, item, namespace);
                // 是否允许从远程同步配置，默认位false
                if (enableRemoteSyncConfig) {
                    String content = getServerConfig(poolId, item, namespace, 3000L);
                    cache.setContent(content);
                }
            }
            // 写时复制
            Map<String, CacheData> copy = new HashMap<String, CacheData>(cacheMap);
            copy.put(key, cache);
            cacheMap = copy;
        }
        log.info("[{}] [subscribe] {}", agent.getName(), key);
        return cache;
    }

    public CacheData getCache(String poolId, String item, String namespace) {
        if (StringUtils.isBlank(poolId) || StringUtils.isBlank(item) || StringUtils.isBlank(namespace)) {
            throw new IllegalArgumentException();
        }
        return cacheMap.get(GroupKeyUtil.getKey(poolId, item, namespace));
    }

    /**
     * 从服务端拉取更新配置
     *
     * @param poolId      线程池id
     * @param item        项目
     * @param namespace   命名空间
     * @param readTimeout 请求超时事件
     * @return 服务端线程池配置
     * @throws TpmProcessException
     */
    public String getServerConfig(String poolId, String item, String namespace, long readTimeout) throws TpmProcessException {
        if (StringUtils.isBlank(item)) {
            item = Constants.DEFAULT_GROUP;
        }

        HttpResult result = null;
        try {
            List<String> params = null;
            if (StringUtils.isBlank(namespace)) {
                params = Arrays.asList("poolId", poolId, "item", item);
            } else {
                params = Arrays.asList("poolId", poolId, "item", item, "namespace", namespace);
            }
            // /v1/cs/configs
            result = agent.httpGet(Constants.CONFIG_CONTROLLER_PATH, null, params, agent.getEncode(), readTimeout);
        } catch (IOException e) {
            String message = String.format(
                    "[%s] [sub-server] get server config exception, poolId=%s, item=%s, namespace=%s", agent.getName(),
                    poolId, item, namespace);
            log.error(message, e);
            throw new TpmProcessException(TpmProcessException.SERVER_ERROR, e);
        }

        switch (result.code) {
            case HttpURLConnection.HTTP_OK:
                return result.content;
            case HttpURLConnection.HTTP_NOT_FOUND:
                return null;
            case HttpURLConnection.HTTP_CONFLICT: {
                log.error(
                        "[{}] [sub-server-error] get server config being modified concurrently, poolId={}, item={}, "
                                + "namespace={}", agent.getName(), poolId, item, namespace);
                throw new TpmProcessException(TpmProcessException.CONFLICT,
                        "data being modified, poolId=" + poolId + ",item=" + item + ",namespace=" + namespace);
            }
            case HttpURLConnection.HTTP_FORBIDDEN: {
                log.error("[{}] [sub-server-error] no right, poolId={}, item={}, namespace={}", agent.getName(), poolId,
                        item, namespace);
                throw new TpmProcessException(result.code, result.content);
            }
            default: {
                log.error("[{}] [sub-server-error]  poolId={}, item={}, namespace={}, code={}", agent.getName(), poolId,
                        item, namespace, result.code);
                throw new TpmProcessException(result.code,
                        "http error, code=" + result.code + ",poolId=" + poolId + ",item=" + item + ",namespace=" + namespace);
            }
        }
    }

    private String null2defaultItem(String item) {
        return (null == item) ? Constants.DEFAULT_GROUP : item.trim();
    }

    /**
     * 每 3000 个配置分配一个长轮询线程
     */
    public void checkConfigInfo() {
        // 获取监听器配置的数量
        int listenerSize = cacheMap.size();
        // 每 3000 个配置需要新增一个长轮询线程
        // 计算出需要执行的长轮询任务数量
        int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());
        // 如果长轮询数量大于当前正在执行的长轮询数量
        if (longingTaskCount > currentLongingTaskCount) {
            for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
                // 通过 executorService 执行长轮询任务（立即执行）
                executorService.execute(new LongPollingRunnable(i));
            }
            // 刷新当前执行的长轮询数量
            currentLongingTaskCount = longingTaskCount;
        }
    }

    /**
     * 找出需要更新的配置
     *
     * @param cacheDataList           存放所有注册监听的线程池（@ThreadPoolManager、DynamicThreadPoolWrap）的 datum
     * @param inInitializingCacheList 用于存放本次轮询新增的配置信息，在入参的时候为空
     * @return 需要更新的配置集合：threadPoolId + projectId + namespace
     */
    private List<String> checkUpdateDataIds(Collection<CacheData> cacheDataList, List<String> inInitializingCacheList) {
        StringBuilder sb = new StringBuilder();
        // 循环拼接当前 cacheDataList 中所有的集合
        for (CacheData cacheData : cacheDataList) {
            sb.append(cacheData.getPoolId()).append(WORD_SEPARATOR);
            sb.append(cacheData.getItem()).append(WORD_SEPARATOR);
            sb.append(cacheData.getMd5()).append(WORD_SEPARATOR);
            sb.append(cacheData.getNamespace()).append(LINE_SEPARATOR);

            // 判断当前的线程池是否是第一次被监控
            if (cacheData.isInitializing()) {
                // 如果是程序启动后第一次被监控
                // 将 threadPoolId + projectId + nameSpace 拼接起来 放到 inInitializingCacheList 集合中
                // 表示当前配置已经被长轮询监听了
                inInitializingCacheList.add(GroupKeyUtil.getKey(cacheData.getPoolId(), cacheData.getItem(), cacheData.getNamespace()));
            }
        }
        // 判断当前集合中有没有新增的线程池配置
        // 如果 inInitializingCacheList 集合不为空（之前没有被监听，新增了线程池配置），不需要挂起当前请求
        boolean isInitializingCacheList = !inInitializingCacheList.isEmpty();
        // 请求服务端获取需要更新的配置信息
        return checkUpdateConfigStr(sb.toString(), isInitializingCacheList);
    }

    /**
     * 请求服务端获取需要更新的配置信息，不会返回null
     *
     * @param probeUpdateString       所有需要监听的所有线程 threadPoolId + projectId + md5 + namespace
     * @param isInitializingCacheList 本次是否有新加入的需要被监控的线程池 true 是
     * @return 需要更新的线程池集合：threadPoolId + projectId + namespace
     */
    public List<String> checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) {

        // 参数是当前所有线程池配置的 MD5 信息
        List<String> params = Arrays.asList(Constants.PROBE_MODIFY_REQUEST, probeUpdateString);

        List<String> headers = new ArrayList<String>(2);
        headers.add(LONG_PULLING_TIMEOUT);
        headers.add("" + timeout);

        // 如果添加了新的初始化缓存数据，告诉服务器不需要长轮询
        if (isInitializingCacheList) {
            headers.add(LONG_PULLING_TIMEOUT_NO_HANGUP);
            headers.add("true");
        }

        // 如果配置是空的直接返回
        if (StringUtils.isBlank(probeUpdateString)) {
            return Collections.emptyList();
        }

        try {
            // 防止服务端处理超时，客户端需要再30s的基础上延长一段等待时间
            long readTimeoutMs = timeout + (long) Math.round(timeout >> 1);

            // 长轮询接口
            // 请求到服务端（1、检查到当前md5异常，会响应；2、29.5s后服务端自动响应；3、长轮询期间服务端配置变化会响应）
            // 路径：/tpm/v1/configs/listener
            HttpResult result = agent.httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", headers, params,
                    agent.getEncode(), readTimeoutMs);
            if (HttpURLConnection.HTTP_OK == result.code) {
                setHealthServer(true);
                return parseUpdateDataIdResponse(result.content);
            } else {
                setHealthServer(false);
                log.error("[{}] [check-update] get changed dataId error, code: {}", agent.getName(), result.code);
            }
        } catch (Exception e) {
            e.printStackTrace();
            setHealthServer(false);
            log.error("[check-update] get changed dataId exception. error message", e);
            throw new TpmProcessException(e.getMessage());
        }
        return Collections.emptyList();
    }

    /**
     * 解析出需要更新的线程池，不会返回null
     *
     * @param response 服务端响应的需要更新的线程池：threadPoolId + projectId + mad5 + namespace
     * @return 需要更新的线程池集合：threadPoolId + projectId + namespace
     */
    public List<String> parseUpdateDataIdResponse(String response) {
        if (StringUtils.isBlank(response)) {
            return Collections.emptyList();
        }

        try {
            response = URLDecoder.decode(response, "UTF-8");
        } catch (Exception e) {
            log.error(">>> [" + agent.getName() + "] [polling-resp] decode modifiedDataIdsString error", e);
        }

        List<String> updateList = new LinkedList();
        for (String dataIdAndGroup : response.split(LINE_SEPARATOR)) {
            if (!StringUtils.isEmpty(dataIdAndGroup)) {
                String[] keyArr = dataIdAndGroup.split(WORD_SEPARATOR);
                if (keyArr.length == 3) {
                    String poolId = keyArr[0];
                    String item = keyArr[1];
                    String namespace = keyArr[2];
                    updateList.add(GroupKeyUtil.getKey(poolId, item, namespace));
                    log.info(">>> [{}] [polling-resp] config changed. threadPoolId={}, projectId={}, namespace={}", agent.getName(), poolId, item, namespace);
                } else {
                    log.error(">>> [{}] [polling-resp] invalid dataIdAndGroup error {}", agent.getName(), dataIdAndGroup);
                }
            }
        }
        return updateList;
    }

    @SuppressWarnings("PMD.ThreadPoolCreationRule")
    public ClientWorker(final HttpAgent agent, final ConfigFilterChainManager configFilterChainManager, final Properties properties) {
        this.agent = agent;
        this.configFilterChainManager = configFilterChainManager;

        // 初始化参数
        init(properties);

        // 循环判断当前需要注册监听的线程池数量，启动对应的长轮询线程监听服务端配置
        // 构建 executor
        this.executor = Executors.newScheduledThreadPool(1, r -> {
            Thread t = new Thread(r);
            t.setName("com.viknix.tpm.client.Worker." + agent.getName());
            // 设置为守护线程
            t.setDaemon(true);
            return t;
        });

        // 获取逻辑核心数，如6核心12线程，那么返回的是12
        int threadSize = Runtime.getRuntime().availableProcessors();
        // 构建 executorService
        this.executorService = Executors.newScheduledThreadPool(threadSize, r -> {
            Thread t = new Thread(r);
            t.setName("com.viknix.tpm.client.Worker.longPolling." + agent.getName());
            // 设置为守护线程
            t.setDaemon(true);
            return t;
        });

        // 维持长轮询，监听服务端线程池配置
        this.executor.scheduleWithFixedDelay(() -> {
            try {
                // 发起长轮询请求，监听服务端配置
                checkConfigInfo();
            } catch (Throwable e) {
                log.error("[sub-check] rotate check error", e);
            }
        }, 1L, 10L, TimeUnit.MILLISECONDS);
    }

    private void init(Properties properties) {
        timeout = Math.max(NumberUtils.toInt(properties.getProperty(PropertyKeyConst.CONFIG_LONG_POLL_TIMEOUT),
                Constants.CONFIG_LONG_POLL_TIMEOUT), Constants.MIN_CONFIG_LONG_POLL_TIMEOUT);

        taskPenaltyTime = NumberUtils.toInt(properties.getProperty(PropertyKeyConst.CONFIG_RETRY_TIME), Constants.CONFIG_RETRY_TIME);

        enableRemoteSyncConfig = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.ENABLE_REMOTE_SYNC_CONFIG));
    }

    class LongPollingRunnable implements Runnable {
        private int taskId;

        public LongPollingRunnable(int taskId) {
            this.taskId = taskId;
        }

        /**
         * 长轮询任务监听服务端配置变化
         */
        @Override
        public void run() {

            // 存放所有的 CacheData
            Collection<CacheData> cacheDatas = new ArrayList<CacheData>();
            // 用于存放本次轮询新增的配置信息
            // 注意：第一轮的时候不会执行长轮询，当 inInitializingCacheList 为 0，没有新增配置的时候才会执行长轮询
            List<String> inInitializingCacheList = new ArrayList();
            try {
                // 切分 cacheMap
                // 将所有的 cacheMap 放到 cacheDataList 中
                for (CacheData cacheData : cacheMap.values()) {
                    if (cacheData.getTaskId() == taskId) {
                        cacheDatas.add(cacheData);
                    }
                }

                // 找出需要更新的配置：poolId + projectId + namespace
                List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);

                // 如果集合为空就不会执行更新
                for (String groupKey : changedGroupKeys) {
                    String[] keys = GroupKeyUtil.parseKey(groupKey);
                    String poolId = keys[0];
                    String item = keys[1];
                    String namespace = keys[2];
                    String content = null;
                    try {
                        // 从服务端拉取更新的配置
                        content = getServerConfig(poolId, item, namespace, 3000L);
                        CacheData cacheData = getCache(poolId, item, namespace);
                        // 这里转一下 PoolParameterInfo 防止 json字符串顺序出现问题
                        String poolContent = ContentUtil.getPoolContent(JSON.parseObject(content, PoolParameterInfo.class));
                        // 设置content，同时设置md5
                        cacheData.setContent(poolContent);
                        log.info(">>> [data-received] namespace :: {}, item :: {}, poolId :: {}, md5 :: {}", namespace, item, poolId, cacheData.getMd5());
                    } catch (Exception e) {
                        // ignore
                        log.info(">>> [data-received] Description Failed to pull the configuration information from the server : {}", content);

                        String message = String.format(
                                "[%s] [get-update] get changed config exception. dataId=%s, item=%s, namespace=%s, content=%s",
                                agent.getName(), poolId, item, namespace, content);
                        log.error(message, e);
                    }
                }

                // 循环检查被监听的 cacheData 中md5配置是否有变化
                for (CacheData cacheData : cacheDatas) {
                    if (!cacheData.isInitializing() || inInitializingCacheList
                            .contains(GroupKeyUtil.getKey(cacheData.getPoolId(), cacheData.getItem(), cacheData.getNamespace()))) {
                        cacheData.checkListenerMd5();
                        // 设置当前 cacheData 已经不是第一次被监控
                        cacheData.setInitializing(false);
                    }
                }

                inInitializingCacheList.clear();
                // 循环执行当前任务
                executorService.execute(this);
            } catch (Throwable e) {

                // If the rotation training task is abnormal, the next execution time of the task will be punished
                log.error("longPolling error : ", e);
                executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS);
            }
        }
    }

}
